package com.hulk.boot.zookeeper.autoconfigure;

import java.nio.charset.Charset;
import java.util.List;

import org.apache.curator.framework.CuratorFramework;
import org.apache.curator.framework.CuratorFrameworkFactory;
import org.apache.curator.framework.CuratorFrameworkFactory.Builder;
import org.apache.curator.framework.api.ACLProvider;
import org.apache.curator.retry.ExponentialBackoffRetry;
import org.apache.zookeeper.ZooDefs;
import org.apache.zookeeper.data.ACL;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.autoconfigure.condition.ConditionalOnClass;
import org.springframework.boot.autoconfigure.condition.ConditionalOnMissingBean;
import org.springframework.boot.context.properties.ConfigurationProperties;
import org.springframework.boot.context.properties.EnableConfigurationProperties;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;

import com.google.common.base.Strings;
import com.hulk.boot.zookeeper.endpoint.ZookeeperEndpoint;
import com.hulk.boot.zookeeper.factory.CuratorFactory;
import com.hulk.boot.zookeeper.health.ZookeeperHealthIndicator;
import com.hulk.boot.zookeeper.nested.NestedZookeeperServers;

@Configuration
@ConditionalOnClass({ org.apache.zookeeper.ZooKeeper.class })
@EnableConfigurationProperties(ZookeeperProperties.class)
public class ZookeeperAutoConfiguration {

	@Autowired
	private ZookeeperProperties zkp;

	private static final String default_namespace = "hulk_path";

	@Bean
	@ConditionalOnMissingBean
	public CuratorFactory curatorFactory() {
		return new CuratorFactory(this.curatorFramework());

	}

	@Bean
	@ConfigurationProperties(prefix = "endpoints.zookeeper", ignoreUnknownFields = false)
	public ZookeeperEndpoint zookeeperEndpoint() {
		return new ZookeeperEndpoint();
	}

	@Bean
	public ZookeeperHealthIndicator zookeeperHealthIndicator() {
		return new ZookeeperHealthIndicator();
	}

	private CuratorFramework curatorFramework() {

		if (this.isUseNestedZookeeper()) {
			NestedZookeeperServers.getInstance().startServerIfNotStarted(zkp.getNestedPort(), zkp.getNestedDataDir());
		}
		Builder builder = CuratorFrameworkFactory.builder().connectString(zkp.getServerLists())
				.retryPolicy(new ExponentialBackoffRetry(zkp.getBaseSleepTimeMilliseconds(), zkp.getMaxRetries(),
						zkp.getMaxSleepTimeMilliseconds()))
				.namespace(Strings.isNullOrEmpty(zkp.getNamespace()) ? default_namespace : zkp.getNamespace());
		if (0 != zkp.getSessionTimeoutMilliseconds()) {
			builder.sessionTimeoutMs(zkp.getSessionTimeoutMilliseconds());
		}
		if (0 != zkp.getConnectionTimeoutMilliseconds()) {
			builder.connectionTimeoutMs(zkp.getConnectionTimeoutMilliseconds());
		}
		if (!Strings.isNullOrEmpty(zkp.getDigest())) {
			builder.authorization("digest", zkp.getDigest().getBytes(Charset.forName("UTF-8")))
					.aclProvider(new ACLProvider() {

						@Override
						public List<ACL> getDefaultAcl() {
							return ZooDefs.Ids.CREATOR_ALL_ACL;
						}

						@Override
						public List<ACL> getAclForPath(final String path) {
							return ZooDefs.Ids.CREATOR_ALL_ACL;
						}
					});
		}
		CuratorFramework client = builder.build();

		client.start();
		try {
			client.getZookeeperClient().blockUntilConnectedOrTimedOut();
		} catch (InterruptedException e) {
			e.printStackTrace();
			throw new RuntimeException(e);
		}

		return client;

	}

	/**
	 * 判断是否需要开启内嵌Zookeeper.
	 * 
	 * @return 是否需要开启内嵌Zookeeper
	 */
	private boolean isUseNestedZookeeper() {
		return -1 != zkp.getNestedPort() && !Strings.isNullOrEmpty(zkp.getNestedDataDir());
	}

}
